/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.performance;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.Vector;

import cgl.narada.transport.sslHttpBase.LinkManager;
import cgl.narada.util.ByteUtilities;
import cgl.narada.util.Log;
import cgl.narada.util.webserver.WebServer;

/**
 * A module that notifies subscribers when there is network performance data to
 * be sent and interpret results passed back from those subscribers.  It contains
 * 1 thread which will notify subscribers when measurement data needs to be send.
 *
 * Currently it will measure round trip latency, jitter, throughput, and messages/
 * second.
 *
 * To get the values, getStats() will return a StatCollection instance with
 * the following metrics:
 *
 * <subscriber name>:jitter
 * <subscriber name>:latency
 * <subscriber name>:throughput - Estimated maximum throughput for the link.
 * <subscriber name>:<metric name> - User specified per second metrics.
 *
 * Configuration Parameters:
 * (a) Measurement %
 * (b) Min Status Interval
 * (c) Max Status Interval
 *
 * @author  John Yin
 * @version 1.0
 */

public class PerformanceMeasurement {

    // Maximum % of throughput the measurement data can be.
    private int  NUM_PING_MEASUREMENTS = 10;
    private static final float MAX_MEASUREMENT_TO_THROUHGPUT_PERCENT = (float)0.02;
    private static final byte[] BANDWIDTH_PAYLOAD = new byte[2000];
    private static final byte[] MEASUREMENT_PAYLOAD = new byte[0];

    private long m_maxStatusInterval = 5 * 60 * 1000;
    private long m_minStatusInterval = 30 * 1000;

    private Vector            m_listeners = new Vector();
    private Hashtable         m_aggregatePSValues = new Hashtable();
    private StatCollection    m_stats     = new StatCollection();
    private MeasurementThread m_thread = new MeasurementThread();
    private WebServer         m_webServer = new WebServer();
    private int               m_webServerPort = 8080;

    private LinkManager    m_channelManager = null;

    /**
     * ------------------------------------------------------------------------
     * Create a default performance Measurement module with 100 samples database
     * per metric.
     */
    public PerformanceMeasurement(LinkManager manager) {
        // Enough to hold 5 minutes worth of samples for the per second
        // metrics.  At 200 ms resolution interval, it generates 5 samples/second
        // for 5 minutes = 5 * 60 * 5 = 1500.
        m_stats.setMaxSamples(1500);
        m_channelManager = manager;
    }

    /**
     * ------------------------------------------------------------------------
     * Sets the web display port.  Default is 8080
     */
    public void setWebInterfacePort(int port) {
        m_webServerPort = port;
    }

    /**
     * ------------------------------------------------------------------------
     */
    public long getMinStatusInterval() {
        return m_minStatusInterval;
    }

    /**
     * ------------------------------------------------------------------------
     */
    public void setMinStatusInterval(long interval) {
        m_minStatusInterval = interval;
    }

    /**
     * ------------------------------------------------------------------------
     */
    public long getMaxStatusInterval() {
        return m_maxStatusInterval;
    }
    /**
     * ------------------------------------------------------------------------
     */
    public void setMaxStatusInterval(long interval) {
        m_maxStatusInterval = interval;
    }

    /**
     * ------------------------------------------------------------------------
     * @return The statistics collection.
     */
    public StatCollection getStatCollection() {
        return m_stats;
    }

    /**
     * ------------------------------------------------------------------------
     */
    public double getStat(String subscriberName,String metric,int numSamples,byte type) {
        return m_stats.getMetricStat(subscriberName+":"+metric,numSamples,type);
    }

    /**
     * ------------------------------------------------------------------------
     */
    public Vector getSamples(String subscriberName,String metric,int numSamples) {
        return m_stats.getSamples(subscriberName+":"+metric,numSamples);
    }

    /**
     * ------------------------------------------------------------------------
     */
    public double getStat(String subscriberName,String metric,
                          long starttime,
                          long endtime,
                          byte type) {
        return m_stats.getMetricStat(subscriberName+":"+metric,starttime,endtime,type);
    }

    /**
     * ------------------------------------------------------------------------
     * @param numSamples The max number of samples in the database.
     */
    public void setSampleBuffer(int numSamples) {
        m_stats.setMaxSamples(numSamples);
    }

    /**
     * ------------------------------------------------------------------------
     * Start the periodic measurement service.
     */
    public void start() {
        m_thread.start();
        m_webServer.setMapping("monitoring/display",
                               new PerformanceRenderer("/monitoring/display",m_channelManager));
        m_webServer.setMapping("monitoring/config",
                               new ConfigCGI("/monitoring/config",this));
        try {
            m_webServer.setPort(m_webServerPort);
            m_webServer.goServer();
            Log.log("PerformanceMeasurement","Display page @ /monitoring/display on port "+m_webServerPort);
            Log.log("PerformanceMeasurement","Display page @ /monitoring/config on port "+m_webServerPort);
        } catch (Exception e) {
            Log.err("PerformanceMeasurement","Unable to start web server",e);
        }
    }

    /**
     * ------------------------------------------------------------------------
     * Stop the periodic measurement service.
     */
    public void stop() {
        m_thread.destroy();
        m_webServer.destroy();
    }

    /**
     * ------------------------------------------------------------------------
     * This will track X values / second type of metric.  Invoke this function
     * everytime there is a change the value.   The time range will be calculated
     * from the first time this method is called for a particular subscriberName/
     * metricName combination to the current measured time.
     *
     * @param subscriberName A unique subscriber name equal to
     *                       PerformanceMeasurementListener.getName();
     * @param metricName The name of the metric ("msg/second", ...)
     * @param value      The value to update the total with (will do += )
     */
    public void updatePSMetric(String subscriberName,String metricName,
                               double value) {
        // Because this function may be called many times and the sample history
        // database is limited in size, we aggregate values here.  The
        // MeasurementThread will aggregate total value for a period and
        // put it into the statistics database to save space.
        // For example, if this method is called once for every message sent,
        // then it will overflow the statistics database easily.
        String name = subscriberName+":"+metricName;
        Double d = (Double)m_aggregatePSValues.get(name);
        if (d == null) {
            d = new Double(value);
        } else {
            d = new Double(value + d.doubleValue());
        }
        m_aggregatePSValues.put(name,d);
    }

    /**
     * ------------------------------------------------------------------------
     * This method should be invoked whenever the transport/measurement subscriber
     * receives a measurement message that was sent out via the
     * PerformanceMeasurementListener.performanceMessagesAvailable().
     *
     * Response messages are of the format:
     * Byte 0: 1
     * Byte 1: SeqNum reply to
     * Byte 2-9: Echo of the ping sent time
     * Byte 10-17: The reception time.
     *
     * @param  subscriberName The unique name of the measurement subscriber
     * @param  measurementMessage The message that was received.
     * @return A vector of response messages(ByteArrayOutputStream) that the
     *         subscriber should deliver
     *         back to the PerformanceMeasurement object that sent the messages.
     * @exception IllegalArgumentException the measurementMessage format is
     *            wrong.
     */
    public Vector getResponse(String subscriberName,byte[] measurementMessage)
        throws IllegalArgumentException
    {
        if (measurementMessage.length != 10 + MEASUREMENT_PAYLOAD.length &&
            measurementMessage.length != (10+BANDWIDTH_PAYLOAD.length)) {
            throw new IllegalArgumentException("Incorrect measurement message size: "+
                                               measurementMessage.length+" should be 10");
        }

        byte seqNum = measurementMessage[1];
        long receiveTime = System.currentTimeMillis();

        Vector v = new Vector();
        ByteArrayOutputStream stream = new ByteArrayOutputStream();
        v.addElement(stream);

        try {
            stream.write(1);
            // Write the ping start time.
            stream.write(measurementMessage,1,9);
            // Write the reception time.
            stream.write(ByteUtilities.getBytes(receiveTime));
        } catch (IOException e) {
            Log.err("PerformanceMeasurement","IO error occurred",e);
        }
        return v;
    }

    /**
     * ------------------------------------------------------------------------
     * This method should be invoked when the measurement sender receives a
     * response message from other side.
     *
     * @param subscriberName The unique name from which reply is for.  Same as the
     *             PerformanceMeasurementListener.getName().
     * @param results The byte stream containing the reply to the measurement
     *             messages sent.
     * @exception IllegalArgumentException results has invalid format.
     */
    public void interpretResults(String subscriberName,
                                 byte[] results) throws IllegalArgumentException {
        if (results[0] != 1) {
            throw new IllegalArgumentException("Invalid result.  Not response type");
        }

        if (results.length != 18 &&
            results.length != (18 + BANDWIDTH_PAYLOAD.length)) {
            throw new IllegalArgumentException("Invalid result.  18 bytes expected but "+
                                               results.length+" received");
        }

        int seqNum = results[1];
        byte[] sentTimeBytes = new byte[8];
        byte[] receiveTimeBytes = new byte[8];
        System.arraycopy(results,2,sentTimeBytes,0,8);
        System.arraycopy(results,10,receiveTimeBytes,0,8);
        long sentTime = ByteUtilities.getLong(sentTimeBytes);
        // Compensate 10 ms for the sent measurement interval required.
        long receiveTime = ByteUtilities.getLong(receiveTimeBytes)-10;
        long currentTime = System.currentTimeMillis();
        int  rt = (int)(receiveTime % 1000000);

/*System.out.println("["+seqNum+"] receiveTime="+receiveTime);
if (seqNum != 0) {
    Vector sample = m_stats.getSamples(subscriberName+":_arrivalTime",1);
    Double s = (Double)sample.elementAt(0);
    int transmissionTime = (rt - (int)s.doubleValue());
    System.out.println("["+seqNum+"->"+(seqNum-1)+"] "+transmissionTime);
}*/
        if (seqNum == NUM_PING_MEASUREMENTS-1) {
            // Now calculate the bandwidth
            Vector sample = m_stats.getSamples(subscriberName+":_arrivalTime",1);
            Double s = (Double)sample.elementAt(0);
            int transmissionTime = (rt - (int)s.doubleValue());
            double estBandwidth = BANDWIDTH_PAYLOAD.length / (double)transmissionTime;
            m_stats.addMetricValue(subscriberName+":throughput",estBandwidth);
//System.out.println("---------------------------------------");
        }

        // Calculate the temporary values.
        m_stats.addMetricValue(subscriberName+":_roundTrip",(currentTime - sentTime));
        // Only need keep a resolution of 8 digits
        m_stats.addMetricValue(subscriberName+":_arrivalTime",rt);

        // Not store the values for round trip and jitter real values.
        if (seqNum == NUM_PING_MEASUREMENTS-2) {
            double avgLatency = m_stats.getMetricStat(subscriberName+":_roundTrip",NUM_PING_MEASUREMENTS-1,StatCollection.AVG_STAT);
            double jitter     = m_stats.getMetricStat(subscriberName+":_arrivalTime",NUM_PING_MEASUREMENTS-1,StatCollection.JITTER_STAT);
            m_stats.addMetricValue(subscriberName+":latency",avgLatency);
            m_stats.addMetricValue(subscriberName+":jitter",jitter);
        }
    }

    /**
     * ------------------------------------------------------------------------
     * Add a subscriber to the performance measurement service.
     */
    public void addMeasurementListener(PerformanceMeasurementListener l) {
        Enumeration e = m_listeners.elements();
        while(e.hasMoreElements()) {
            ListenerRecord r = (ListenerRecord)e.nextElement();
            if (r.m_listener.equals(l)) {
                return;
            }
        }
        ListenerRecord r = new ListenerRecord(l);
        if (r.m_statusInterval < m_minStatusInterval) {
            r.m_statusInterval = m_minStatusInterval;
        } else if (r.m_statusInterval > m_maxStatusInterval) {
            r.m_statusInterval = m_maxStatusInterval;
        }
        m_listeners.addElement(r);
    }

    /**
     * ------------------------------------------------------------------------
     */
    public void removeMeasurementListener(PerformanceMeasurementListener l) {

        Enumeration e = m_listeners.elements();
        while(e.hasMoreElements()) {
            ListenerRecord r = (ListenerRecord)e.nextElement();
            if (r.m_listener.equals(l)) {
                m_listeners.removeElement(r);
                return;
            }
        }
    }

    /**
     * ======================================================================
     */
    class MeasurementThread extends Thread {

        private boolean m_stopped = false;
        private long    m_sentIntervalStart = 0;
        private int     m_sentBytes         = 0;

        /**
         * ------------------------------------------------------------------------
         */
        public void run() {
            while(!m_stopped) {
                try { Thread.currentThread().sleep(200); } catch (Exception e) { }
                Enumeration e = m_listeners.elements();
                long currentTime = System.currentTimeMillis();
                while(e.hasMoreElements()) {
                    ListenerRecord r = (ListenerRecord)e.nextElement();
                    long interval = r.m_statusInterval;
                    if (r.m_lastStatustime == 0) {
                        r.m_lastStatustime = System.currentTimeMillis();
                    } else if ((currentTime - r.m_lastStatustime) >= interval) {
                        measurementInterval(r);
                    } else {
                    }
                }

                e = m_aggregatePSValues.keys();
                while(e.hasMoreElements()) {
                    String key = (String)e.nextElement();
                    Double v   = (Double)m_aggregatePSValues.remove(key);
                    m_stats.addMetricValue(key,v.doubleValue());
                }
            }
        }

        /**
         * ------------------------------------------------------------------------
         * Perform a measurement step.
         */
        private void measurementInterval(ListenerRecord r) {
            long currentTime = System.currentTimeMillis();
            if (m_sentBytes == 0) {
                m_sentIntervalStart = currentTime;
            }

            Vector data = generatePingData();
            r.m_listener.performanceMessagesAvailable(data);
            r.m_lastStatustime = currentTime;

            // Update bytes sent during this interval
            Enumeration e = data.elements();
            while(e.hasMoreElements()) {
                ByteArrayOutputStream baos = (ByteArrayOutputStream)e.nextElement();
                m_sentBytes += baos.size();
            }

            // Time to check to see if we are sending out too much data.
            long diff = currentTime - m_sentIntervalStart;
            if (diff >= 1000) {
                float kbPS = (float)m_sentBytes / diff;

                throttleStatusInterval(r,kbPS);
                m_sentIntervalStart = 0;
                m_sentBytes = 0;
            }
        }

        /**
         * ------------------------------------------------------------------------
         * Based on this subscriber's network health, throttle the status interval.
         */
        private void throttleStatusInterval(ListenerRecord r,float kbPS) {
            String name = r.m_listener.getName();
            try {
                // Get the last sample.
                double throughput = getStat(name,"throughput",1,StatCollection.MAX_STAT);
                float  percent = kbPS / (float)throughput;
                if (percent > MAX_MEASUREMENT_TO_THROUHGPUT_PERCENT) {
                    r.m_statusInterval *= 2;
                } else {
                    r.m_statusInterval /= 2;
                }

                if (r.m_statusInterval > m_maxStatusInterval) {
                    r.m_statusInterval = m_maxStatusInterval;
                } else if (r.m_statusInterval < m_minStatusInterval) {
                    r.m_statusInterval = m_minStatusInterval;
                }
            } catch (IllegalArgumentException e) {
            }
        }

        /**
         * ------------------------------------------------------------------------
         * Each message has the following format:
         *
         * byte 0: Type [ 0 = request, 1 = response ]
         * byte 1: Seq. #
         * byte 2-9: client ping time.
         */
        private Vector generatePingData() {
            Vector v = new Vector();

            for(int i=0;i < NUM_PING_MEASUREMENTS;i++) {
                ByteArrayOutputStream stream = new ByteArrayOutputStream(10);
                stream.write(0); // request
                stream.write(i);
                try {
                    stream.write(ByteUtilities.getBytes(System.currentTimeMillis()));
                    if (i == NUM_PING_MEASUREMENTS-1) {
                        stream.write(BANDWIDTH_PAYLOAD);
                    } else if (MEASUREMENT_PAYLOAD.length != 0) {
                        stream.write(MEASUREMENT_PAYLOAD);
                    }
                } catch (IOException e) {
                    Log.err("PerformanceMeasurement","Unable to generate ping data",e);
                }
                v.addElement(stream);
            }
            return v;
        }

        /**
         * ------------------------------------------------------------------------
         */
        public void destroy() {
            if (!m_stopped) {
                m_stopped = true;
                interrupt();
            }
        }
    }
}


/**
 * ======================================================================
 */
class ListenerRecord {
    PerformanceMeasurementListener m_listener;
    long m_lastStatustime = 0;
    long m_statusInterval = 0;

    ListenerRecord(PerformanceMeasurementListener l) {
        m_listener = l;
        m_statusInterval = m_listener.getStatusInterval();
    }
}
